package com.lpssfxy.online.model

import com.lpssfxy.online.entities.{MongoConfig, ProductRecs, Recommendation, StreamRecs}
import com.lpssfxy.online.utils.ConnHelper
import com.lpssfxy.online.utils.AppUtils.{MAX_SIM_PRODUCTS_NUM, MAX_USER_RATING_NUM, MONGODB_RATING_COLLECTION, MONGO_DB, MONGO_URI, PRODUCT_RECS, REDIS_HOST, REDIS_PASSWORD, REDIS_PORT, STREAM_RECS}
import com.mongodb.casbah.Imports.MongoClientURI
import com.mongodb.casbah.MongoClient
import com.mongodb.casbah.commons.MongoDBObject
import org.apache.log4j.Logger
import org.apache.spark.SparkContext
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.dstream.DStream
import redis.clients.jedis.{DefaultJedisClientConfig, Jedis}

import scala.collection.JavaConversions._

/**
 * 该对象包含了处理流式计算的主要逻辑，用于实时处理用户评分数据并生成商品推荐列表。
 */
object StreamComputeProcessor {
  private val logger = Logger.getLogger(getClass)

  /**
   * 创建广播变量，将商品相似度矩阵广播到各个节点。
   *
   * @param spark Spark会话
   * @param sc Spark上下文
   * @return 广播变量，包含商品相似度矩阵
   */
  def createProductRecsBroadcast(spark: SparkSession, sc: SparkContext)(implicit mongoConfig: MongoConfig):
  Broadcast[scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]]] = {
    try {
      import spark.implicits._
      // 加载数据，相似度矩阵，广播出去
      val simProductsMatrix = spark.read
        .option("uri", mongoConfig.uri)
        .option("collection", PRODUCT_RECS)
        .format("com.mongodb.spark.sql")
        .load()
        .as[ProductRecs]
        .rdd
        // 为了后续查询相似度方便，把数据转换成map形式
        .map { item =>
          (item.productId, item.recs.map(x => (x.productId, x.score)).toMap)
        }
        .collectAsMap()
      // 定义广播变量
      sc.broadcast(simProductsMatrix)
    } catch {
      case e: Exception =>
        logger.error("[CUSTOM]Error creating product recs broadcast", e)
        throw e
    }
  }

  /**
   * 运行评分处理器，处理评分流数据并生成实时推荐列表。
   *
   * @param ratingStream 评分流，数据格式为 (userId, productId, score, timestamp)
   * @param simProductMatrixBC 广播变量，包含商品相似度矩阵
   */
  def runRatingProcessor(ratingStream: DStream[(Int, Int, Double, Int)], simProductMatrixBC: Broadcast[scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]]]): Unit = {
    ratingStream.foreachRDD { rdd =>
      rdd.foreachPartition { partition =>
        // 分区内创建连接（序列化安全）
        val redisClientConfig = DefaultJedisClientConfig.builder()
          .password(REDIS_PASSWORD)
          .build()
        val jedis = new Jedis(REDIS_HOST, REDIS_PORT, redisClientConfig)
        val mongoClient = MongoClient(MongoClientURI(MONGO_URI))
        implicit val mongoConfig = MongoConfig(MONGO_URI, MONGO_DB)

        try {
          partition.foreach {
            case (userId, productId, score, timestamp) =>
              // 业务逻辑（与原代码一致）
              println(">>>>>>>>>rating data coming!>>>>>>>>>>>>>>>>>>")

              // 1. 从redis里取出当前用户的最近评分，保存成一个数组Array[(productId, score)]
              val userRecentlyRatings = getUserRecentlyRatings(MAX_USER_RATING_NUM, userId, ConnHelper.jedis)

              // 2. 从相似度矩阵中获取当前商品最相似的商品列表，作为备选列表，保存成一个数组Array[productId]
              val candidateProducts = getTopSimProducts(MAX_SIM_PRODUCTS_NUM, productId, userId, simProductMatrixBC.value)

              // 3. 计算每个备选商品的推荐优先级，得到当前用户的实时推荐列表，保存成 Array[(productId, score)]
              val streamRecs = computeProductScore(candidateProducts, userRecentlyRatings, simProductMatrixBC.value)

              // 4. 存入mongodb
              saveDataToMongoDB(userId, streamRecs, mongoClient)

          }
        } finally {
          // 分区处理完成后必须关闭连接（重要！避免资源泄漏）
          jedis.close()
          mongoClient.close()
        }
      }
    }
  }

  /**
   * 将推荐结果更新或保存到MongoDB。
   * 首先根据用户ID删除原有的推荐记录，然后插入新的推荐记录。
   * @param userId       用户的ID。
   * @param mongoClient  mongoClient客户端
   * @param streamRecs   推荐结果数组，每个元素是一个元组，包含商品ID和评分。
   */
  def saveDataToMongoDB(userId: Int, streamRecs: Array[(Int, Double)], mongoClient: MongoClient)(implicit mongoConfig: MongoConfig): Unit = {
    val recs = streamRecs.map { case (productId, score) => Recommendation(productId, score) }.toSeq
    val streamRecsObj = StreamRecs(userId, recs)
    val streamRecsCollection = mongoClient(mongoConfig.db)(STREAM_RECS)
    streamRecsCollection.findAndRemove(MongoDBObject("userId" -> userId))
    streamRecsCollection.insert(MongoDBObject("userId" -> userId, "recs" -> streamRecsObj.recs.map(x => MongoDBObject("productId" -> x.productId, "score" -> x.score))))
  }

  /**
   * 从redis里获取最近num次评分。
   * @param num    要获取的评分数量
   * @param userId 用户ID
   * @param jedis  Redis连接对象
   * @return      包含最近评分的数组，格式为 (productId, score)
   */
  def getUserRecentlyRatings(num: Int, userId: Int, jedis: Jedis): Array[(Int, Double)] = {
    try {
      // 从redis中用户的评分队列里获取评分数据，list键名为uid:USERID，值格式是 PRODUCTID:SCORE
      jedis.lrange("userId:" + userId.toString, 0, num)
        .map { item =>
          val attr = item.split("\\:")
          (attr(0).trim.toInt, attr(1).trim.toDouble)
        }
        .toArray
    } catch {
      case e: Exception =>
        logger.error(s"[CUSTOM]Error getting user recently ratings for user $userId", e)
        throw e
    }
  }

  /**
   * 获取当前商品的相似列表，并过滤掉用户已经评分过的，作为备选列表。
   *
   * @param num         要获取的相似商品数量
   * @param productId   当前商品ID
   * @param userId      用户ID
   * @param simProducts 商品相似度矩阵
   * @return            包含备选商品ID的数组
   */
  def getTopSimProducts(num: Int,
                        productId: Int,
                        userId: Int,
                        simProducts: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]])
                       (implicit mongoConfig: MongoConfig): Array[Int] = {
    // 从广播变量相似度矩阵中拿到当前商品的相似度列表
    val allSimProducts = simProducts(productId).toArray

    // 获得用户已经评分过的商品，过滤掉，排序输出
    val ratingCollection = ConnHelper.mongoClient(mongoConfig.db)(MONGODB_RATING_COLLECTION)
    println(s"当前用户的ID：${userId}")
    val ratingExist = ratingCollection.find(MongoDBObject("userId" -> userId))
      .toArray
      .map { item => // 只需要productId
        item.get("productId").toString.toInt
      }
    // 从所有的相似商品中进行过滤
    allSimProducts.filter(x => !ratingExist.contains(x._1))
      .sortWith(_._2 > _._2)
      .take(num)
      .map(x => x._1)
  }

  /**
   * 计算每个备选商品的推荐得分。
   *
   * @param candidateProducts   备选商品ID数组
   * @param userRecentlyRatings 用户最近评分数组，格式为 (productId, score)
   * @param simProducts         商品相似度矩阵
   * @return 包含推荐商品ID和得分的数组，格式为 (productId, score)
   */
  def computeProductScore(candidateProducts: Array[Int],
                          userRecentlyRatings: Array[(Int, Double)],
                          simProducts: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]]): Array[(Int, Double)] = {
    try {
      // 定义一个长度可变数组ArrayBuffer，用于保存每一个备选商品的基础得分，(productId, score)
      val scores = scala.collection.mutable.ArrayBuffer[(Int, Double)]()
      // 定义两个map，用于保存每个商品的高分和低分的计数器，productId -> count
      val increMap = scala.collection.mutable.HashMap[Int, Int]()
      val decreMap = scala.collection.mutable.HashMap[Int, Int]()

      // 遍历每个备选商品，计算和已评分商品的相似度
      for (candidateProduct <- candidateProducts; userRecentlyRating <- userRecentlyRatings) {
        // 从相似度矩阵中获取当前备选商品和当前已评分商品间的相似度
        val simScore = getProductsSimScore(candidateProduct, userRecentlyRating._1, simProducts)
        if (simScore > 0.6) {  // 过滤出相似度大于0.6的商品
          // 按照公式进行加权计算，得到基础评分
          scores += ((candidateProduct, simScore * userRecentlyRating._2))
          if (userRecentlyRating._2 > 3) {
            increMap(candidateProduct) = increMap.getOrDefault(candidateProduct, 0) + 1
          } else {
            decreMap(candidateProduct) = decreMap.getOrDefault(candidateProduct, 0) + 1
          }
        }
      }
      // 根据公式计算所有的推荐优先级，首先以productId做groupBy
      scores.groupBy(_._1).map {
          case (productId, scoreList) =>
            (productId, scoreList.map(_._2).sum / scoreList.length + log(increMap.getOrDefault(productId, 1)) - log(decreMap.getOrDefault(productId, 1)))
        }
        // 返回推荐列表，按照得分排序
        .toArray
        .sortWith(_._2 > _._2)
    } catch {
      case e: Exception =>
        logger.error("[CUSTOM] Error computing product score", e)
        throw e
    }
  }

  /**
   * 获取两个商品之间的相似度得分。
   *
   * @param product1 商品ID 1
   * @param product2 商品ID 2
   * @param simProducts 商品相似度矩阵
   * @return   两个商品之间的相似度得分
   */
  private def getProductsSimScore(product1: Int, product2: Int,
                                  simProducts: scala.collection.Map[Int, scala.collection.immutable.Map[Int, Double]]): Double = {
    try {
      simProducts.get(product1) match {
        case Some(sims) => sims.get(product2) match {
          case Some(score) => score // 匹配到了直接返回score
          case None => 0.0          // 没有匹配到直接返回0.0
        }
        case None => 0.0
      }
    } catch {
      case e: Exception =>
        logger.error(s"[CUSTOM]Error getting product sim score between $product1 and $product2", e)
        0.0
    }
  }

  /**
   * 自定义log函数，以N为底，实现了换底公式。
   *
   * @param m 输入值
   * @return 以N为底的对数结果
   */
  private def log(m: Int): Double = {
    try {
      val N = 10
      math.log(m) / math.log(N)
    } catch {
      case e: Exception =>
        logger.error(s"[CUSTOM]Error computing log for value $m", e)
        0.0
    }
  }
}